package com.shujia.spark.optimize

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02Coalesce {
  def main(args: Array[String]): Unit = {
    /**
      * coalesce  重分区
      *
      * 改变rdd并行度，如果资源充足，可以提高并行度提高任务执行速度
      *
      * 保证每一个task处理的数据再50到128mb不等
      */

    val conf: SparkConf = new SparkConf()
      .setAppName("app")
      .setMaster("local[1]")
    val sc: SparkContext = new SparkContext(conf)

    //产生小文件
//    sc
//      .textFile("data/students.txt")
//    //进行重分区 会产生shuffle
//      .repartition(20)
//      .saveAsTextFile("data/repartition")

    val rdd: RDD[String] = sc.textFile("data/repartition")

    println("rdd:" + rdd.getNumPartitions)
    /**
      * repartition 会产生shuffle
      *
      */
    val rdd2: RDD[String] = rdd.repartition(3)

    println("rdd2:" + rdd2.getNumPartitions)

    /**
      * 增加分区必须产生shuffle
      * 减少分区的时候可以不产生shuffle
      */
    /**
      * 合并小文件
      *
      * 保证每一个task处理的数据量在100m左右
      */

    //使用coalesce减少分区，可以减少分区的时间，并且不产生分区

    val DRdd: RDD[String] = rdd.coalesce(2,false)

    println("DRdd:"+DRdd.getNumPartitions)


    //如果使用分区必须要使用shuffle
    val ARdd: RDD[String] = rdd.coalesce(200,true)

    println("ARdd:"+ARdd.getNumPartitions)
  }
}
